package com.lzwk.app.dwd;

/**
 * @Author: CC
 * @Date: 2022/1/5 17:57
 */

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lzwk.app.function.UrlUtil;
import com.lzwk.config.JobConfig;
import com.lzwk.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class BaseLogShunt {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        JobConfig prod = new JobConfig("prod", env);

        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(prod.earthLogTopic, prod.earthLogGroupId, prod.earthKafkaServers));

        //脏数据流
        OutputTag<String> outputTag = new OutputTag<String>("Dirty") {
        };
        //ods
        OutputTag<String> odsTag = new OutputTag<String>("ods") {
        };
        //wk_h5
        OutputTag<String> wkH5Tag = new OutputTag<String>("wk_h5") {
        };
        //wk_server
        OutputTag<String> wkServerTag = new OutputTag<String>("wk_server") {
        };
        //wk_app
        OutputTag<String> wkAppTag = new OutputTag<String>("wk_app") {
        };
        //lh_app
        OutputTag<String> lhAppTag = new OutputTag<String>("lh_app") {
        };
        //ocpx
        OutputTag<String> ocpxTag = new OutputTag<String>("ocpx") {
        };
        //lx_h5
        OutputTag<String> lxH5Tag = new OutputTag<String>("lx_h5") {
        };
        //others
        OutputTag<String> othersTag = new OutputTag<String>("others") {
        };

        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    //解析日志
                    String log = jsonObject.getString("log");
                    JSONObject resultJsonObject = UrlUtil.parse(log);
                    ctx.output(odsTag, resultJsonObject.toJSONString());
                    out.collect(resultJsonObject);
                } catch (Exception e) {
                    ctx.output(outputTag, value);
                }
            }
        });

        SingleOutputStreamOperator<String> pageDS = jsonObjDS.process(new ProcessFunction<JSONObject, String>() {
                                                                          @Override
                                                                          public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                                                                              // 根据不同spm写入dwd
                                                                              String spm = value.getString("spm");
                                                                              if (spm == null) {
                                                                                  spm = "";
                                                                              }
                                                                              if (spm.startsWith("wka")) {
                                                                                  ctx.output(wkAppTag, value.toJSONString());
                                                                              } else if (spm.startsWith("wks")) {
                                                                                  ctx.output(wkServerTag, value.toJSONString());
                                                                              } else if (spm.startsWith("wk")) {
                                                                                  ctx.output(wkH5Tag, value.toJSONString());
                                                                              } else if (spm.startsWith("oc")) {
                                                                                  ctx.output(ocpxTag, value.toJSONString());
                                                                              } else if (spm.startsWith("lha")) {
                                                                                  ctx.output(lhAppTag, value.toJSONString());
                                                                              } else if (spm.startsWith("lx")) {
                                                                                  ctx.output(lxH5Tag, value.toJSONString());
                                                                              } else {
                                                                                  ctx.output(othersTag, value.toJSONString());
                                                                              }
                                                                          }
                                                                      }
        );
        //打印脏数据
        DataStream<String> dirtyDS = jsonObjDS.getSideOutput(outputTag);
        DataStream<String> odsDS = jsonObjDS.getSideOutput(odsTag);
        DataStream<String> wkH5DS = pageDS.getSideOutput(wkH5Tag);
        DataStream<String> wkServerDS = pageDS.getSideOutput(wkServerTag);
        DataStream<String> wkAppDS = pageDS.getSideOutput(wkAppTag);
        DataStream<String> lhAppDS = pageDS.getSideOutput(lhAppTag);
        DataStream<String> ocpxDS = pageDS.getSideOutput(ocpxTag);
        DataStream<String> lxH5DS = pageDS.getSideOutput(lxH5Tag);
        DataStream<String> othersDS = pageDS.getSideOutput(othersTag);

//        dirtyDS.print("odsDS>>>>>>>>>>>")
//        odsDS.print("odsDS>>>>>>>>>>>");
//        wkH5DS.print("wkH5DS>>>>>>>>>>>");
//        wkServerDS.print("wkServerDS>>>>>>>>>>>");
//        wkAppDS.print("wkAppDS>>>>>>>>>>>");
//        ocpxDS.print("ocpxDS>>>>>>>>>>>");
//        lxH5DS.print("lxH5DS>>>>>>>>>>>");
//        othersDS.print("othersDS>>>>>>>>>>>");
//        pageDS.print("Page>>>>>>>>>>>");

        dirtyDS.addSink(MyKafkaUtil.getKafkaProducer("dirty_earth_log", prod.earthKafkaServers));
        odsDS.addSink(MyKafkaUtil.getKafkaProducer("ods_earth_log_clean", prod.earthKafkaServers));
        wkH5DS.addSink(MyKafkaUtil.getKafkaProducer("dwd_wk_log", prod.earthKafkaServers));
        wkServerDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_wks_log", prod.earthKafkaServers));
        wkAppDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_wka_log", prod.earthKafkaServers));
        lhAppDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_lha_log", prod.earthKafkaServers));
        ocpxDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_oc_log", prod.earthKafkaServers));
        lxH5DS.addSink(MyKafkaUtil.getKafkaProducer("dwd_lx_log", prod.earthKafkaServers));
        othersDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_others_log", prod.earthKafkaServers));
        env.execute("earth_log_shunt");
    }
}
